import java.util.*;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.WritableComparator;

public class SpecialK implements WritableComparable<SpecialK> {// implements WritableComparable <SpecialK> {
  private Long prefix = new Long(0);
//  private Text prefix = new Text();
  private int offset;

  public long getPrefix() {
//  public Text getPrefix() {
    return prefix;
  }

  public int getOffset() {
    return offset;
  }

  public void setPrefix(long prefix) {
//  public void setPrefix(Text prefix) {
//this.prefix.set(prefix);
    this.prefix = prefix;
  }

//  public void setPrefix(String prefix) {
//    this.prefix.set(prefix);
//  }

  public void setOffset(int offset) {
    this.offset = offset;
  }

  public int hashCode() {
//    return this.prefix;
    return prefix.hashCode();
  }

  public void readFields(DataInput in) throws IOException {
    prefix = in.readLong();
//    prefix.readFields(in);
    offset = in.readInt();
  }

  public void write(DataOutput out) throws IOException {
    out.writeLong(prefix);
//    prefix.write(out);
    out.writeInt(offset);
  }
  
  public SpecialK() {
  }

  public SpecialK(Long prefix, int offset) {
    this.prefix=prefix;
//    this.prefix.set(prefix);
    this.offset=offset;
  }
  
/*  public SpecialK(String prefix, int offset) {
    this.prefix.set(prefix);
    this.offset=offset;
  }*/

  public int compareTo(SpecialK that) {
    if (this.getOffset() < that.getOffset()) {
      return -1;
    } else if (this.getOffset() > that.getOffset()) {
      return 1;
    } else {
      return 0;
    }
  }

  @Override
  public String toString() {
    return (prefix + " " + offset);
  }

  //GroupComparator Groups same prefix which comes sorted from SortComparator
  public static class GroupComparator extends WritableComparator {
    public GroupComparator() {
      super(SpecialK.class);
    }

    public int compare (byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
      try {
//Code for Text
/*        int n1 = WritableUtils.decodeVIntSize(b1[s1]);
        int n2 = WritableUtils.decodeVIntSize(b2[s2]);
        int result = WritableComparator.compareBytes(b1, s1+n1, l1-n1, b2, s2+n2, l2-n2);*/
        long thisp = readLong(b1, s1);
        long thatp = readLong(b2, s2);

//        System.out.println("GroupComparator " + thisp + " " + thatp);

        if (thisp < thatp) {
          return -1;
        } else if (thisp > thatp) {
          return 1;
        } else {
          return 0;
/*          int thiso = readInt(b1, s1+l1);
          int thato = readInt(b1, s1+l1);
        
          if (thiso < thato) {
            return -1;
          } else if (thiso > thato) {
            return 1;
          } else {
            return 0;
          }*/
        }

/*        if (thisp<thatp) {
          return -1;
        } else if (thisp>thatp) {
          return 1;
        } else {
          if (thiso < thato) {
            return -1;
          } else if (thiso > thato) {
            return 1;
          } else {
            return 0;
          }
        }*/
      } catch (ArrayIndexOutOfBoundsException e) {
        System.out.println("ERROR ARRAY OUT OF BOUNDS");
        return 0;
      }
    }
  }

  //SortComparator sorts by both fields
  public static class SortComparator extends WritableComparator {
    public SortComparator() {
      super(SpecialK.class);
    }

    public int compare (byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
      try {
//Code for Text
/*        int n1 = WritableUtils.decodeVIntSize(b1[s1]);
        int n2 = WritableUtils.decodeVIntSize(b2[s2]);
        int result = WritableComparator.compareBytes(b1, s1+n1, l1-n1, b2, s2+n2, l2-n2);*/
        long thisp = readLong(b1, s1);
        long thatp = readLong(b2, s2);

//        System.out.println("SortComparator " + thisp + " " + thatp);

        if (thisp < thatp) {
          return -1;
        } else if (thisp > thatp) {
          return 1;
        } else {
          int thiso = readInt(b1, s1+l1);
          int thato = readInt(b2, s2+l2);
        
          if (thiso < thato) {
            return -1;
          } else if (thiso > thato) {
            return 1;
          } else {
            return 0;
          }
        }
      } catch (ArrayIndexOutOfBoundsException e) {
        System.out.println("ERROR ARRAY OUT OF BOUNDS");
        return 0;
      }
    }
  }

  static {
    // register this comparator
    WritableComparator.define(SpecialK.class, new GroupComparator());
    WritableComparator.define(SpecialK.class, new SortComparator());
  }

  }
